package com.huawei.test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Hello world!
 */
public class App {
    public static void main(String[] args) throws Exception {
            // 1. 创建执行环境
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            // 2. 读取数据源（此处使用文本文件作为数据源）
            //DataStream<String> text = env.readTextFile("src/main/resources/words.txt");
            DataStream<String> text = env.readTextFile(args[0]);

            // 3. 对数据流进行转换处理
            DataStream<Tuple2<String, Integer>> counts = text
                    .flatMap(new Tokenizer())
                    .keyBy(value -> value.f0)
                    .reduce(new SumReducer());

            // 4. 结果输出
            counts.print();

            // 5. 启动作业
            env.execute("Flink WordCount Example");
        }

        // 分词器 FlatMapFunction
        public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
                // 将输入行拆分为单词，并输出 (word, 1) 的 Tuple2
                for (String word : value.split("\\s+")) {
                    if (word.length() > 0) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            }
        }
        // 累加器 ReduceFunction
        public static final class SumReducer implements ReduceFunction<Tuple2<String, Integer>> {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) {
                // 将两个 (word, count) 的 Tuple2 中 count 值相加
                return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
            }
        }
    }



